package com.qhx.rtkcloud.tcp;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.net.Socket;
import java.util.Arrays;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TcpExchangeUnit {

    public static final ThreadLocal<TcpExchangeUnit> TCP_EXCHANGE_UNIT_THREAD_LOCAL = new ThreadLocal<>();
    private static final Logger log = LoggerFactory.getLogger(TcpExchangeUnit.class);

    private BlockingQueue<byte[]> readQueue;
    private BlockingQueue<byte[]> writeQueue;
    private OutputStream outputStream = null;
    private InputStream inputStream = null;
    private Socket client = null;

    private boolean connectFreedom = true;

    private boolean writeFreedom = true;

    private TcpExchangeUnit partner;

    private int authType;

    /**
     * writeFreedom为true 启动/连接断开时，获取该锁，连接成功释放
     *  writeFreedom为false 启动时获取partner的锁
     */
    public Object connectLock = new Object();

    /**
     * 可以作为扩展，写到子类里
     */
    private String mount;

    public int getAuthType() {
        return authType;
    }

    public void setAuthType(int authType) {
        this.authType = authType;
    }

    public boolean isWriteFreedom() {
        return writeFreedom;
    }

    public void setWriteFreedom(boolean writeFreedom) {
        this.writeFreedom = writeFreedom;
    }

    public TcpExchangeUnit getPartner() {
        return partner;
    }

    public void setPartner(TcpExchangeUnit partner) {
        this.partner = partner;
    }

    public boolean isConnectFreedom() {
        return connectFreedom;
    }

    public void setConnectFreedom(boolean connectFreedom) {
        this.connectFreedom = connectFreedom;
    }

    public Socket getClient() {
        return client;
    }

    public void setClient(Socket client) {
//        closeSocket();
//        readThread.interrupt();
//        writeThread.interrupt();
        this.client = client;
        try {
            this.outputStream = this.client.getOutputStream();
            this.inputStream = this.client.getInputStream();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public TcpExchangeUnit(BlockingQueue writeQueue) {
        this.writeQueue = writeQueue;
    }

    private Thread readThread;
    private Thread writeThread;

    public void start() {
       readThread = new Thread(() -> readData(), "read:" + this.getClient().getInetAddress() + ":" + this.client.getPort() + "-" + this.getClient().getLocalPort());
       readThread.start();
       writeThread = new Thread(() -> writeData(), "write:" + this.getClient().getInetAddress() + ":" + this.getClient().getPort() + "-" + this.getClient().getLocalPort());
       writeThread.start();
    }

    public String getMount() {
        return mount;
    }

    public void setMount(String mount) {
        this.mount = mount;
    }

    public BlockingQueue<byte[]> getReadQueue() {
        return readQueue;
    }

    public void setReadQueue(BlockingQueue<byte[]> readQueue) {
        this.readQueue = readQueue;
    }

    public BlockingQueue<byte[]> getWriteQueue() {
        return writeQueue;
    }

    public void setWriteQueue(BlockingQueue<byte[]> writeQueue) {
        this.writeQueue = writeQueue;
    }

    public OutputStream getOutputStream() {
        return outputStream;
    }

    public void setOutputStream(OutputStream outputStream) {
        this.outputStream = outputStream;
    }

    public InputStream getInputStream() {
        return inputStream;
    }

    public void setInputStream(InputStream inputStream) {
        this.inputStream = inputStream;
    }

    private void closeSocket() {
        try {
            if (client != null && !client.isClosed()) {
                client.close();
            }

            if (outputStream != null) {
                outputStream.close();
            }

            if (inputStream != null) {
                inputStream.close();
            }

        } catch (IOException exception) {
            exception.printStackTrace();
        }
    }

    protected byte[] writeHook(byte[] bytes) {
        return bytes;
    }

    protected byte[] readHook(byte[] bytes) {
        return bytes;
    }


    private void reflect(Object object) {
        Class clazz = object.getClass();

        Field field = null;
        try {
            field = clazz.getDeclaredField("socket");
            field.setAccessible(true);
            Socket socket = (Socket) field.get(object);
            log.info(socket.getInetAddress() + ":" + socket.getPort() + ";" + socket.getLocalAddress() + ":" + socket.getLocalPort());
        } catch (NoSuchFieldException e) {
            throw new RuntimeException(e);
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        }



    }


    private void writeData() {
        byte[] info = null;
        try {
            while (!Thread.currentThread().isInterrupted() &&  (info = readQueue.take()) != null) {
                log.info(client.getInetAddress() + ":" + client.getPort() + "," + client.getLocalAddress() + ":" + client.getLocalPort() + ": write data:" +  new String((info)));
                reflect(outputStream);
                outputStream.write(writeHook(info));
                outputStream.flush();
            }
        } catch (InterruptedException e) {
            log.error("interrupted while writing data", e);

        } catch (IOException e) {
            log.error("ioexception while writing data", e);
            closeSocket();
            TCP_EXCHANGE_UNIT_THREAD_LOCAL.set(this);
            throw new RuntimeException("" + client.getInetAddress().getHostAddress() + ":" + client.getPort(), e);
        } catch (Exception e) {
            log.error("exception while writing data", e);
        }
        }

    private void readData() {
        byte[] bytes = new byte[1024];
        int length = 0;
        try {
            while (true && !Thread.currentThread().isInterrupted()) {
                reflect(outputStream);
                length  = inputStream.read(bytes);
                if (length == -1) {
                    log.error("receice -1");
                    throw new IOException();
                }
                log.info(client.getInetAddress() + ":" + client.getPort() + "," + client.getLocalAddress() + ":" + client.getLocalPort() + ": read data:" + new String((Arrays.copyOfRange(bytes, 0, length))));

                    if (writeFreedom || (partner.client != null && !partner.client.isClosed())) {
                        writeQueue.offer(readHook(Arrays.copyOfRange(bytes, 0, length)));
                    }

            }

        } catch (IOException exception) {
            log.error("ioexception read writing data", exception);
            closeSocket();
            TCP_EXCHANGE_UNIT_THREAD_LOCAL.set(this);
            throw new RuntimeException((mount == null ?"" :mount) + ":" + client.getInetAddress().getHostAddress() + ":" + client.getPort(), exception);
        } catch (Exception e) {
            log.error("exception read writing data", e);
            e.printStackTrace();
        }
    }


}

